package com.eascs.web.o2o.mq.test;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class RabbitConsumer {
	private static final String TASK_QUEUE_NAME = "task_queue";

	static ExecutorService es = Executors.newFixedThreadPool(2);

	public static void main(String[] args) throws IOException, TimeoutException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("172.29.12.201");
		factory.setPort(5672);
		factory.setUsername("guest");
		factory.setPassword("kaibin");
		Connection connection = factory.newConnection(es);
		Channel channel = connection.createChannel(2);

		channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
		System.out.println("Worker1  Waiting for messages");

		// 每次从队列获取的数量
		channel.basicQos(2);

		com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
				String message = new String(body, "UTF-8");
				Date date = new Date();
				SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				String dateStr = sdf.format(date);
				System.out.println(Thread.currentThread().getName() + "Worker1  Received '" + message + "' ==" + dateStr);
				try {
					// throw new Exception();
					doWork(message);
				} finally {
					System.out.println("Worker1 Done");
					channel.basicAck(envelope.getDeliveryTag(), false);
					// channel.basicNack(envelope.getDeliveryTag(), false,
					// false);
				}
			}
		};
		// 消息消费完成确认
		channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
	}

	private static void doWork(String task) {
		try {
			Thread.sleep(2000); // 暂停1秒钟
		} catch (InterruptedException _ignored) {
			Thread.currentThread().interrupt();
		}
	}
}
